package com.tsf.hbase;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;

/**
 * hbase实现样例
 *
 * @author taoninigbo
 * @since 1.0
 */
public class HBaseTemplate extends HBaseAccessor implements HBaseOperations {

    /**
     * 构造一个{@code HBaseTemplate}实例.
     *
     * @param hbaseClient 一个{@code HBaseClient}的实例
     */
    public HBaseTemplate(HBaseClient hbaseClient) {
        super(hbaseClient);
    }

    /**
     * 测试表中是否存在列(由Get指定)。
     * <p>
     * 如果Get匹配一个或多个键，则返回true;如果不匹配，则返回false。
     * <p>
     * 这是一个服务器端调用，因此可以防止将任何数据传输到客户机。
     *
     * @param tableName 表名称
     * @param get       用于对单行执行Get操作
     * @return 如果指定的Get匹配一个或多个键，则为true;如果不匹配，则为false
     */
    @Override
    public boolean exists(String tableName, Get get) {
        Table table = getTable(tableName);
        if (table != null) {
            try {
                return table.exists(get);
            } catch (IOException e) {
                LOG.error("Test for the existence of columns in the table error.", e);
            } finally {
                close(table);
            }
        }
        return false;
    }

    /**
     * 测试表中是否存在列(由Gets指定)。
     * <p>
     * 这将返回一个布尔值数组。如果相关的Get匹配一个或多个键，则每个值为true，如果不匹配，则为false。
     * <p>
     * 这是一个服务器端调用，因此可以防止将任何数据传输到客户机。
     *
     * @param tableName 表名称
     * @param gets      用于对单行执行Get操作
     * @return 布尔值的数组。如果指定的Get匹配一个或多个键，则为True;如果不匹配，则为false。
     */
    @Override
    public boolean[] existsAll(String tableName, List<Get> gets) {
        Table table = getTable(tableName);
        if (table != null) {
            try {
                return table.existsAll(gets);
            } catch (IOException e) {
                LOG.error("Test for the existence of columns in the table error.", e);
            } finally {
                close(table);
            }
        }
        return new boolean[]{};
    }

    /**
     * 保存一批数据
     *
     * @param tableName 表名称
     * @param puts      待保存的一批数据
     * @return 如果保存成功返回'true'，否则返回'false'
     */
    @Override
    public boolean put(String tableName, List<Put> puts) {
        BufferedMutator mutator = null;
        try {
            mutator = connection().getBufferedMutator(TableName.valueOf(tableName));
            mutator.mutate(puts);
            mutator.flush();
        } catch (Exception e) {
            LOG.error("Save data to hbase server error, {}", e.getMessage());
            return false;
        } finally {
            if (mutator != null) {
                try {
                    mutator.close();
                } catch (IOException e) {
                    LOG.warn("Closed BufferedMutator error. {}", e.getMessage());
                }
            }
        }
        return true;
    }

    /**
     * 保存一条数据
     *
     * @param tableName 表名称
     * @param put       待保存的一条数据
     * @return 如果保存成功返回'true'，否则返回'false'
     */
    @Override
    public boolean put(String tableName, Put put) {
        Table table;
        if ((table = getTable(tableName)) != null) {
            try {
                table.put(put);
                return true;
            } catch (IOException e) {
                return false;
            } finally {
                close(table);
            }
        } else {
            return false;
        }
    }

    /**
     * 获取一条数据
     *
     * @param tableName 表名称
     * @param get       用于对单行执行Get操作
     * @return 查询的单行结果
     */
    @Override
    public Result get(String tableName, Get get) {
        Table table = getTable(tableName);
        if (table != null) {
            try {
                return table.get(get);
            } catch (IOException e) {
                LOG.error("Extracts certain cells from a given row error.", e);
            } finally {
                close(table);
            }
        }
        return null;
    }

    /**
     * 获取多条数据
     *
     * @param tableName 表名称
     * @param gets      用于对单行执行Get操作
     * @return 查询的单行结果
     */
    @Override
    public Result[] get(String tableName, List<Get> gets) {
        Table table = getTable(tableName);
        if (table != null) {
            try {
                return table.get(gets);
            } catch (IOException e) {
                LOG.error("Extracts certain cells from a given row error.", e);
            } finally {
                close(table);
            }
        }
        return new Result[]{};
    }

    /**
     * 删除指定的单元格/行。
     *
     * @param tableName 表名称
     * @param delete    指定要删除哪些内容的对象。
     */
    @Override
    public void delete(String tableName, Delete delete) {
        Table table = getTable(tableName);
        if (table != null) {
            try {
                table.delete(delete);
            } catch (IOException e) {
                LOG.error("Deletes the specified cells/row error.", e);
            } finally {
                close(table);
            }
        }
    }

    /**
     * 批量删除指定的单元格/行。
     *
     * @param tableName 表名称
     * @param deletes   要删除的内容列表。List通过这个方法进行修改(特别是它被重新排序，
     *                  因此在列表中插入元素的顺序不能保证执行{@link Delete}的顺序)。
     */
    @Override
    public void delete(String tableName, List<Delete> deletes) {
        Table table = getTable(tableName);
        if (table != null) {
            try {
                table.delete(deletes);
            } catch (IOException e) {
                LOG.error("Deletes the specified cells/row error.", e);
            } finally {
                close(table);
            }
        }
    }

    /**
     * 非分页扫描指定{@code tableName}表中的且只返回指定列{@code columns}的所有数据
     *
     * @param tableName 表名称
     * @param columns   指定返回这些列名的数据，key为列名称，value为列族名称
     * @param startKey  指定这个参数后，会从此位置开始扫描
     * @param endKey    指定这个参数后，会从扫描到此位置结束
     * @return 返回一个hbase的扫描结果
     */
    @Override
    public ResultScanner scanner(String tableName, Map<String, String> columns, byte[] startKey, byte[] endKey) {
        ResultScanner results = null;
        Table table = getTable(tableName);
        if (table != null) {
            try {
                Scan scan = new Scan();
                if (columns != null && !columns.isEmpty()) {
                    for (String columnName : columns.keySet()) {
                        scan.addColumn(Bytes.toBytes(columns.get(columnName)), Bytes.toBytes(columnName));
                    }
                }
                if (startKey != null) {
                    scan.setStartRow(startKey);
                }
                if (endKey != null) {
                    scan.setStopRow(endKey);
                }
                results = table.getScanner(scan);
            } catch (IOException e) {
                LOG.error("Get scanner result error.", e);
            } finally {
                close(table);
            }
        }
        return results;
    }

    /**
     * 非分页扫描指定{@code tableName}表中的所有数据
     *
     * @param tableName 表名称
     * @param startKey  指定这个参数后，会从此位置开始扫描
     * @param endKey    指定这个参数后，会从扫描到此位置结束
     * @return 返回一个hbase的扫描结果
     */
    @Override
    public ResultScanner scanner(String tableName, byte[] startKey, byte[] endKey) {
        return scanner(tableName, null, startKey, endKey, null);
    }

    /**
     * 非分页扫描指定{@code tableName}表中的且符合过滤条件{@code filters}的所有数据
     *
     * @param tableName 表名称
     * @param startKey  指定这个参数后，会从此位置开始扫描
     * @param endKey    指定这个参数后，会从扫描到此位置结束
     * @param filters   过滤条件
     * @return 返回一个hbase的扫描结果
     */
    @Override
    public ResultScanner scanner(String tableName, byte[] startKey, byte[] endKey, FilterList filters) {
        return scanner(tableName, null, startKey, endKey, filters);
    }

    /**
     * 非分页扫描指定{@code tableName}表中的且列族名为{@code families}的所有数据
     *
     * @param tableName 表名称
     * @param families  这个表的所有列族名称
     * @param startKey  指定这个参数后，会从此位置开始扫描
     * @param endKey    指定这个参数后，会从扫描到此位置结束
     * @return 返回一个hbase的扫描结果
     */
    @Override
    public ResultScanner scanner(String tableName, Collection<String> families, byte[] startKey, byte[] endKey) {
        return scanner(tableName, families, startKey, endKey, null);
    }

    /**
     * 非分页扫描指定{@code tableName}表中的且列族名为{@code families}的且符合过滤条件{@code filters}的所有列的数据
     *
     * @param tableName 表名称
     * @param families  这个表的所有列族名称
     * @param startKey  指定这个参数后，会从此位置开始扫描
     * @param endKey    指定这个参数后，会从扫描到此位置结束
     * @param filters   过滤条件
     * @return 返回一个hbase的扫描结果
     */
    @Override
    public ResultScanner scanner(String tableName, Collection<String> families, byte[] startKey, byte[] endKey, FilterList filters) {
        ResultScanner scanner = null;
        Table table = getTable(tableName);
        if (table != null) {
            try {
                Scan scan = new Scan();
                if (families != null) {
                    for (String family : families) {
                        scan.addFamily(Bytes.toBytes(family));
                    }
                }
                if (filters != null) {
                    scan.setFilter(filters);
                }
                if (startKey != null) {
                    scan.setStartRow(startKey);
                }
                if (endKey != null) {
                    scan.setStopRow(endKey);
                }
                scanner = table.getScanner(scan);
            } catch (IOException e) {
                LOG.error("Get scanner result error.", e);
            } finally {
                close(table);
            }
        }
        return scanner;
    }

    /**
     * 分页扫描指定{@code tableName}表中的所有列的数据
     *
     * @param tableName 表名称
     * @param families  这个表的所有列族名称
     * @param pageSize  分页大小
     * @param startKey  指定这个参数后，会从此位置开始查询
     * @param endKey    指定这个参数后，会从扫描到此位置结束
     * @param startTs   过滤数据的起始时间
     * @param endTs     过滤数据的结束时间
     * @return 返回一个hbase的扫描结果
     */
    @Override
    public ResultScanner pageScanner(String tableName, Collection<String> families, int pageSize, byte[] startKey, byte[] endKey, Long startTs, Long endTs) {
        return pageScanner(tableName, families, pageSize, startKey, endKey, startTs, endTs, null);
    }

    /**
     * 分页扫描指定{@code tableName}表中的所有列数据
     *
     * @param tableName 表名称
     * @param families  这个表的所有列族名称
     * @param pageSize  分页大小
     * @param startKey  指定这个参数后，会从此位置开始查询
     * @param endKey    指定这个参数后，会从扫描到此位置结束
     * @param startTs   过滤数据的起始时间
     * @param endTs     过滤数据的结束时间
     * @param filters   过滤条件
     * @return 返回一个hbase的扫描结果
     */
    @Override
    public ResultScanner pageScanner(String tableName, Collection<String> families, int pageSize, byte[] startKey, byte[] endKey, Long startTs, Long endTs, FilterList filters) {
        ResultScanner scanner = null;
        Table table = getTable(tableName);
        if (table != null) {
            try {
                Scan scan = new Scan();
                if (families != null) {
                    for (String family : families) {
                        scan.addFamily(Bytes.toBytes(family));
                    }
                }
                if (filters != null) {
                    filters.addFilter(new PageFilter(pageSize + 1));
                    scan.setCaching(pageSize + 1);
                    scan.setFilter(filters);
                } else {
                    scan.setFilter(new PageFilter(pageSize + 1));
                    scan.setCaching(pageSize + 1);
                }
                if (startKey != null) {
                    scan.setStartRow(startKey);
                }
                if (endKey != null) {
                    scan.setStopRow(endKey);
                }
                if (startTs != null && endTs != null) {
                    scan.setTimeRange(startTs, endTs);
                }
                scanner = table.getScanner(scan);
            } catch (IOException e) {
                LOG.error("Get scanner result error.", e);
            } finally {
                close(table);
            }
        }
        return scanner;
    }
}
